package com.intelligent.system.common.manager;

import com.intelligent.system.common.dto.MessageDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public class SseEmitterManager {
    
    // 存储所有客户端连接，key为用户ID
    private static final Map<Long, SseEmitter> SSE_CACHE = new ConcurrentHashMap<>();
    
    // 默认超时时间：30分钟
    private static final long DEFAULT_TIMEOUT = 30 * 60 * 1000L;
    
    /**
     * 创建SSE连接
     *
     * @param userId 用户ID
     * @return SseEmitter
     */
    public SseEmitter createConnection(Long userId) {
        // 如果已存在连接，先移除
        removeConnection(userId);
        
        // 创建新的SSE连接
        SseEmitter emitter = new SseEmitter(DEFAULT_TIMEOUT);
        
        // 设置超时回调
        emitter.onTimeout(() -> {
            log.info("SSE连接超时，用户ID：{}", userId);
            removeConnection(userId);
        });
        
        // 设置完成回调
        emitter.onCompletion(() -> {
            log.info("SSE连接完成，用户ID：{}", userId);
            removeConnection(userId);
        });
        
        // 设置错误回调
        emitter.onError(throwable -> {
            log.error("SSE连接异常，用户ID：{}", userId, throwable);
            removeConnection(userId);
        });
        
        // 存储连接
        SSE_CACHE.put(userId, emitter);
        log.info("创建SSE连接成功，用户ID：{}", userId);
        
        return emitter;
    }
    
    /**
     * 移除SSE连接
     *
     * @param userId 用户ID
     */
    public void removeConnection(Long userId) {
        SseEmitter emitter = SSE_CACHE.remove(userId);
        if (emitter != null) {
            try {
                emitter.complete();
            } catch (Exception e) {
                log.error("关闭SSE连接异常，用户ID：{}", userId, e);
            }
        }
    }
    
    /**
     * 向指定用户推送消息
     *
     * @param userId 用户ID
     * @param message 消息内容
     * @return 是否推送成功
     */
    public boolean sendMessage(Long userId, MessageDTO message) {
        SseEmitter emitter = SSE_CACHE.get(userId);
        if (emitter == null) {
            log.warn("用户未建立SSE连接，无法推送消息，用户ID：{}", userId);
            return false;
        }
        
        try {
            emitter.send(message);
            log.info("消息推送成功，用户ID：{}，消息ID：{}", userId, message.getId());
            return true;
        } catch (IOException e) {
            log.error("消息推送失败，用户ID：{}，消息ID：{}", userId, message.getId(), e);
            removeConnection(userId);
            return false;
        }
    }
    
    /**
     * 向多个用户推送消息
     *
     * @param userIds 用户ID列表
     * @param message 消息内容
     * @return 成功推送的用户数量
     */
    public int sendMessageToUsers(List<Long> userIds, MessageDTO message) {
        if (userIds == null || userIds.isEmpty()) {
            return 0;
        }
        
        int successCount = 0;
        for (Long userId : userIds) {
            if (sendMessage(userId, message)) {
                successCount++;
            }
        }
        return successCount;
    }
    
    /**
     * 向所有在线用户推送消息
     *
     * @param message 消息内容
     * @return 成功推送的用户数量
     */
    public int broadcastMessage(MessageDTO message) {
        if (SSE_CACHE.isEmpty()) {
            return 0;
        }
        
        int successCount = 0;
        for (Long userId : SSE_CACHE.keySet()) {
            if (sendMessage(userId, message)) {
                successCount++;
            }
        }
        return successCount;
    }
    
    /**
     * 获取当前在线用户数量
     *
     * @return 在线用户数量
     */
    public int getOnlineUserCount() {
        return SSE_CACHE.size();
    }
    
    /**
     * 判断用户是否在线
     *
     * @param userId 用户ID
     * @return 是否在线
     */
    public boolean isUserOnline(Long userId) {
        return SSE_CACHE.containsKey(userId);
    }
} 